﻿using Atomic;
using JT808.Protocol;
using JT808.Protocol.Enums;
using JT808.Protocol.Exceptions;
using JT808.Protocol.Extensions;
using JT808.Protocol.Extensions.JTActiveSafety.MessageBody;
using JT808.Protocol.Extensions.JTActiveSafety.Metadata;
using JT808.Protocol.MessageBody;
using JT808.Protocol.MessagePack;
using JTActiveSafety.Gateway.Abstractions;
using JTActiveSafety.Gateway.Services;
using JTActiveSafety.Gateway.Sessions;
using JTActiveSafety.Protocol;
using JTActiveSafety.Protocol.Extensions;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Buffers;
using System.Buffers.Binary;
using System.Collections.Concurrent;
using System.IO.Pipelines;
using System.Linq;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;

namespace JTActiveSafety.Gateway
{
    public class JTActiveSafetyMessageHandler : IAsyncDisposable
    {
        private Task _processingTask;
        private readonly IKestrelTrace _trace;
        private readonly ConnectionContext _socketConnection;
        // This locks access to all of the below fields
        private readonly object _contextLock = new object();

        private readonly object _sendingLock = new object();

        private readonly IJT808Config _jt808Config;
        private readonly JT808Serializer _jt808serializer;
        private readonly IAttachFileStorageService _attachFileStorageService;
        private readonly IAttachFileRecvService _attachFileRecvService;
        private readonly JTActiveSafetySessionManager _sessionManager;

        private readonly ConcurrentPipeWriter _sender;
        private readonly PipeReader _reader;

        //private JTActiveSafetyTcpSession _tcpSession;
        private ConcurrentDictionary<string, JTActiveSafetySession> _sessions = new ConcurrentDictionary<string, JTActiveSafetySession>(StringComparer.OrdinalIgnoreCase);

        //当前只实现了一个连接对应一个终端Session 
        public JTActiveSafetyMessageHandler(ConnectionContext socketConnection, MemoryPool<byte> pool, IKestrelTrace trace)
        {
            _socketConnection = socketConnection;
            _trace = trace;
            _reader = _socketConnection.Transport.Input;
            // Allow appending more data to the PipeWriter when a flush is pending.
            _sender = new ConcurrentPipeWriter(_socketConnection.Transport.Output, pool, _contextLock);

            using (var cs = ServiceLocator.Instance.CreateScope())
            {
                var serviceProvider = cs.ServiceProvider;

                _jt808Config = serviceProvider.GetService<IJT808Config>();
                _jt808serializer = _jt808Config.GetSerializer();

                //_messageHandler = serviceProvider.GetService<JT808MessageHandler>();

                _attachFileStorageService = serviceProvider.GetService<IAttachFileStorageService>();

                _attachFileRecvService = serviceProvider.GetService<IAttachFileRecvService>();

                _sessionManager = serviceProvider.GetService<JTActiveSafetySessionManager>();
            }
        }

        public void Execute()
        {
            _processingTask = ExecuteAsync();

        }
    
        private async Task ExecuteAsync()
        {
            //_tcpSession = new JTActiveSafetyTcpSession(_socketConnection);
            //_sessionManager.TryAdd(_tcpSession);
            try
            {
                var receiveTask = ProcessRequestsAsync();

                Task.WaitAll(receiveTask);
            }
            catch (Exception ex)
            {
                _trace.RequestProcessingError(_socketConnection.ConnectionId, ex);
            }
            finally
            {
                //_sessionManager.RemoveBySessionId(_tcpSession.SessionID);
            }
        }

        private Task SendAsync(ReadOnlyMemory<byte> data)
        {
            try
            {
                _sender.WriteAsync(data);
                _sender.FlushAsync();
            }
            catch (Exception ex)
            {
                _trace.LogWarning(ex,$"发送数据到终端时出错：{data.ToArray().ToHexString()}");
            }
            return Task.CompletedTask;
        }

        //ProcessRequestsAsync
        private async Task ProcessRequestsAsync(CancellationToken cancellationToken = default)
        {
            try
            {
                while (true)
                {
                    var result = await _reader.ReadAsync(cancellationToken);//注意在调用 PipeReader.ReadAsync 之后始终调用 PipeReader.AdvanceTo
                    var readableBuffer = result.Buffer;
                    var consumed = readableBuffer.Start;
                    var examined = readableBuffer.End;

                    try
                    {
                        if (result.IsCanceled)
                        {
                            break;//读取已取消。您可以不读取现有数据而退出。
                        }

                        if (!readableBuffer.IsEmpty)
                        {
                            ParsePreface(ref readableBuffer, ref consumed, ref examined);
                        }

                        if (result.IsCompleted)
                        {
                            //if (buffer.Length > 0)
                            //{
                            //    // The message is incomplete and there's no more data to process.
                            //    throw new InvalidDataException("Incomplete message.");
                            //}
                            break;//没有更多的数据要处理
                        }
                    }
                    catch (Exception ex)
                    {

                        _trace.RequestProcessingError(_socketConnection.ConnectionId, ex);
                        break;
                    }
                    finally
                    {
                        _reader.AdvanceTo(consumed,examined);//已读取了多少
                        //reader.AdvanceTo(buffer.GetPosition(index));
                        //reader.AdvanceTo(buffer.Start, buffer.End);
                        //reader.AdvanceTo(consumed/*确定消耗的内存量*/, examined/*确定观察到的缓冲区数*/);
                    }
                }
            }
            catch (Exception ex)
            {
                _trace.RequestProcessingError(_socketConnection.ConnectionId, ex);
            }
            finally
            {
                await _reader.CompleteAsync();//.Complete();
            }
        }

        #region ActiveSefety
        /// <summary>
        /// 读取包消息，每次只读取一条808或主动安全附件消息。
        /// </summary>
        /// <param name="buffer">源</param>
        /// <param name="consumed">读取进度</param>
        /// <param name="examined"></param>
        /// <returns></returns>
        private void ParsePreface(ref ReadOnlySequence<byte> buffer, ref SequencePosition consumed, ref SequencePosition examined)
        {
            //读取一个完整包 常见异常及处理方式
            //1.长度不够   终止打包，等待下次读取。
            //2.已找到808或主动安全包头，读取时未找到终止符(808)或长度不够（主动安全）  终止打包，等待下次读取。
            //3.找不到808或主动安全包头，终止打包，等待下次读取。
            var reader = new SequenceReader<byte>(buffer);
            var readLength = 0;
            while (!reader.End)
            {
                //808包最小14位 主动安全附件文件包最小62位
                if (reader.Length < 14)
                {
                    break;
                }
 
                var packType = ExtractActiveSafetyPack(ref reader, _trace);
 
                _trace.LogDebug($"read pack:{_socketConnection.ConnectionId} {_socketConnection.RemoteEndPoint} {packType} {reader.Length}");
                if (packType == 808)
                {
                    if (reader.TryRead808Package(out var _span))
                    {
                        readLength += _span.Length;
                        _trace.LogDebug($"Connection id \"{_socketConnection.ConnectionId}\"[{_socketConnection.RemoteEndPoint}] [808 Parse ===>]: {_span.ToArray().ToHexString()}");

                        Hander808Parse(_span).ConfigureAwait(false);

                        break;
                    }
                }
                else if (packType == 999)
                {
                    ReadActiveSafety:
                    if (reader.TryReadActiveSafetyPackage(out var _span2))
                    {
                        readLength += _span2.Length;

                        _trace.LogDebug($"Connection id \"{_socketConnection.ConnectionId}\"[{_socketConnection.RemoteEndPoint}] [ActiveSafety Parse ===>]:{_span2.ToArray().ToHexString()}");

                        HanderActiveSafetyParse(_span2).ConfigureAwait(false);

                        //主动安全包后面，大概率还是主动安全包   4158=4096+62  
                        if (reader.Remaining >= 4158) {
                            goto ReadActiveSafety;
                        }
                        
                        break;
                    }
                }
                break;
            }
            if (reader.End)
            {
                examined = consumed = buffer.End;
            }
            else
            {
                examined = consumed = reader.Position;//buffer.GetPosition(readLength);
            }
        }
        /// <summary>
        /// 从数据中提取出符合主动安全规范的包（包括3个交互808指令0x1210/0x1211/0x1212 和上传的主动安全附件文件数据包），不符合的数据将被截取并丢弃。
        /// 同时，这也是一个安全检查，服务器对外提供服务不可避免的遇到网络刺探。
        /// </summary>
        /// <param name="reader"></param>
        /// <param name="trace">808：808包  999：主动安全附件包 其他值:无效数据包</param>
        private static int ExtractActiveSafetyPack(ref SequenceReader<byte> reader, IKestrelTrace trace)
        {
            if (reader.TryReadBigEndian(out short sss))
                reader.Rewind(2);
            if (sss == 32274 /*0x7e 0x12*/)
            {
                return 808;
            }

            if (reader.TryReadBigEndian(out int ss))
                reader.Rewind(4);
            if (ss == 808543076 /*0x30 31 63 64*/)
            {
                return 999;
            }

            trace.LogDebug($"收到非808或主动安全包头：{reader.Sequence.ToArray().ToHexString()}");
            if (reader.TryReadTo(out ReadOnlySpan<byte> _span, new byte[] { 0x7E, 0x12 }, false))
            {
                trace.LogDebug($"已剔除数据:{_span.ToArray().ToHexString()}");
                return 0;
            }
            //if (reader.TryReadTo(out ReadOnlySpan<byte> _, new byte[] { 0x7E, 0x92 }, false))
            //{
            //    trace.LogInformation($"已剔除数据:{_span.ToArray().ToHexString()}");
            //    return 0;
            //}
            if (reader.TryReadTo(out ReadOnlySpan<byte> _, new byte[] { 0x30, 0x31, 0x63, 0x64 }, false))
            {
                trace.LogDebug($"已剔除数据:{_span.ToArray().ToHexString()}");
                return 0;
            }

            reader.Advance(reader.Remaining);
            return 0;
        }

        private Task Hander808Parse(ReadOnlySpan<byte> span)
        {
            try
            {
                var package = _jt808serializer.HeaderDeserialize(span);
                _sessionManager.AddOrUpdate(new JTActiveSafetyUdpSession(_socketConnection, package.Header.TerminalPhoneNo));
 
                if(package.Header.MsgId == 0x1210)
                {
                    _ = SendAsync(Msg0x1210(package));
                }
                else if(package.Header.MsgId == 0x1211)
                {
                    _ = SendAsync(Msg0x1211(package));
                }
                else if (package.Header.MsgId == 0x1212)
                {
                    _ = SendAsync(Msg0x1212(package));
                }
            }
            catch (JT808Exception ex)
            {
                var msg = $"[HeaderDeserialize ErrorCode1]:{ ex.ErrorCode},[ReaderBuffer]:{_socketConnection.RemoteEndPoint},{span.ToArray().ToHexString()}";
                _trace.RequestProcessingError(_socketConnection.ConnectionId, new Exception(msg));
            }
            catch (Exception ex)
            {
                _trace.RequestProcessingError(_socketConnection.ConnectionId, ex);
            }
            return Task.CompletedTask;
        }

        private Task HanderActiveSafetyParse(ReadOnlySpan<byte> span)
        {
            try
            {
                var package = JTActiveSafetySerializer.Deserialize(span);
                _ = SendAsync(Msg0x1211_2(package));
                //var package = _jt808serializer.HeaderDeserialize(span);
                //_sessionManager.AddOrUpdate(new JTActiveSafetyUdpSession(_socketConnection, package.Header.TerminalPhoneNo));
                //_sessionManager.TryLink(_tcpSession.TerminalPhoneNo, _tcpSession);
                //_ = _attachFileStorageService.WriteData(_tcpSession.TerminalPhoneNo, span).ConfigureAwait(false);
                //await _msgProducer.WriteData(session.TerminalPhoneNo, span);

                //var _span = ((ushort)0x1299).CreateCustomMsgId(session.TerminalPhoneNo, new JT808_0x1299()
                //{
                //    FileName = Encoding.UTF8.GetString(span.Slice(4, 50).ToArray()),
                //    Offset = BinaryPrimitives.ReadUInt32BigEndian(span.Slice(54, 4)),
                //    Length = BinaryPrimitives.ReadUInt32BigEndian(span.Slice(58, 4)),
                //});
                //var jt808_1299 = _serializer.Serialize(_span).ToArray();
                //var package = _serializer.HeaderDeserialize(jt808_1299);
                //var downData = _messageHandler.Processor(package);
                //session.SendAsync(downData);
            }
            catch (Exception ex)
            {
                _trace.RequestProcessingError(_socketConnection.ConnectionId, ex);
            }
            return Task.CompletedTask;
        }

        private void AddOrUpdate(JTActiveSafetySession session)
        {
            _sessions.AddOrUpdate(session.SessionId, session, (k, v) =>
            {
                v.LastActiveTime = DateTime.Now;
                return v;
            });
        }
        /// <summary>
        /// 平台通用应答
        /// </summary>
        /// <param name="request"></param>
        private byte[] CommonReply(JT808HeaderPackage request)
        {
            if (request.Version == JT808Version.JTT2019)
            {
                byte[] data = _jt808serializer.Serialize(JT808MsgId.平台通用应答.Create_平台通用应答_2019(request.Header.TerminalPhoneNo, new JT808_0x8001()
                {
                    AckMsgId = request.Header.MsgId,
                    JT808PlatformResult = JT808PlatformResult.成功,
                    MsgNum = request.Header.MsgNum
                }));
                return data;
            }
            else
            {
                byte[] data = _jt808serializer.Serialize(JT808MsgId.平台通用应答.Create(request.Header.TerminalPhoneNo, new JT808_0x8001()
                {
                    AckMsgId = request.Header.MsgId,
                    JT808PlatformResult = JT808PlatformResult.成功,
                    MsgNum = request.Header.MsgNum
                }));
                return data;
            }
        }

        /// <summary>
        /// 终端发送附件文件信息列表
        /// </summary>
        /// <param name="request"></param>
        private byte[] Msg0x1210(JT808HeaderPackage request)
        {
            _trace.LogDebug($"0x1210 {request.Header.TerminalPhoneNo}"/* {request.Bodies.ToHexString()}"*/);

            try
            {
                JT808MessagePackReader tmpReader = new JT808MessagePackReader(request.Bodies);
                var jtt1210 = _jt808Config.GetMessagePackFormatter<JT808_0x1210>().Deserialize(ref tmpReader, _jt808Config);
                _attachFileRecvService.CreateAlarmAttachFileInfo(request.Header.TerminalPhoneNo, jtt1210);
                return CommonReply(request);
            }
            catch (Exception ex)
            {
                _trace.LogError(ex, $"处理消息0x1210错误：{request.Header.TerminalPhoneNo} {request.Bodies.ToHexString()}");
            }
            return default;
        }
        /// <summary>
        /// 终端发送附件文件传输
        /// </summary>
        /// <param name="request"></param>
        private byte[] Msg0x1211(JT808HeaderPackage request)
        {
            _trace.LogDebug($"0x1211 {request.Header.TerminalPhoneNo}"/* {request.Bodies.ToHexString()}" */);
            return CommonReply(request);
            //try
            //{

            //    return CommonReply(request);
            //}
            //catch (Exception ex)
            //{
            //    _logger.LogError(ex, $"处理消息0x1211错误：{request.Header.TerminalPhoneNo} {request.Bodies.ToHexString()}");
            //}
            //return default;
        }
        /// <summary>
        /// 自定义消息
        /// </summary>
        /// <param name="span"></param>
        /// <returns></returns>
        private byte[] Msg0x1211_2(JTActiveSafetyPackage package)
        {
            try
            {

                _attachFileStorageService.WriteData(package);
            }
            catch (Exception ex)
            {
                _trace.LogError(ex, $"处理自定义消息0x1211_2错误：{package.FileName} {package.Offset} {package.Length} {package.Bodies.ToHexString()}");
            }

            return default;
        }

        /// <summary>
        /// 终端附件上传完成应答
        /// </summary>
        /// <param name="request"></param>
        private byte[] Msg0x1212(JT808HeaderPackage request)
        {
            _trace.LogDebug($"0x1212 {request.Header.TerminalPhoneNo}"/* {request.Bodies.ToHexString()}" */);

            //Thread.Sleep(1000); //对于1212消息延迟应答

            try
            {
                var tmpReader = new JT808MessagePackReader(request.Bodies);
                var jt1212 = _jt808Config.GetMessagePackFormatter<JT808_0x1212>().Deserialize(ref tmpReader, _jt808Config);

                var jt9212 = new JT808_0x9212() { FileName = jt1212.FileName, FileNameLength = jt1212.FileNameLength };

                var r = _attachFileRecvService.CheckFileIntegrity(jt1212.FileName, out var fileSize, out var missingNodes);
                if (r == CheckFileIntegrityResult.数据错误 || r == CheckFileIntegrityResult.文件已存储成功
                    || r == CheckFileIntegrityResult.找不到对应文件记录 || r == CheckFileIntegrityResult.检查成功_文件完整)
                {
                    goto end;
                }
                else if (r == CheckFileIntegrityResult.打不到文件数据包记录)
                {
                    missingNodes.Add(new Node() { Offset = 0, Length = jt1212.FileSize });
                    _trace.LogDebug($"完整补传记录：{jt1212.FileName} size:{jt1212.FileSize}");
                }
                else if (r == CheckFileIntegrityResult.检查成功_文件不完整)
                {
                    foreach (var item in missingNodes)
                    {
                        _trace.LogDebug($"补传记录：{jt1212.FileName} size:{jt1212.FileSize} offset:{item.Offset} length:{item.Length}");
                    }
                }

                
            end:
                jt9212.DataPackages = missingNodes;
                jt9212.DataPackageCount = (byte)jt9212.DataPackages.Count;
                jt9212.UploadResult = (byte)(jt9212.DataPackages.Count > 0 ? 1 : 0);

                //return _jt808serializer.Serialize(jt9212);
                return _jt808serializer.Serialize(((ushort)0x9212).CreateCustomMsgId(request.Header.TerminalPhoneNo, jt9212));
            }
            catch (Exception ex)
            {
                _trace.LogError(ex, $"处理消息0x1212错误：{request.Header.TerminalPhoneNo} {request.Bodies.ToHexString()}");
                return default;
            }
        }

        //private void Remove(string sessionId)
        //{
        //    if (_sessions.TryRemove(sessionId, out var _))
        //    {

        //    }
        //}
        #endregion


        public async ValueTask DisposeAsync()
        {
            if (_processingTask != null)
            {
                await _processingTask;
            }
        }
    }

    //public abstract class JTActiveSafetyProtocol : IRequestProcessor
    //{
    //    public void Abort(ConnectionAbortedException ex)
    //    {
    //        throw new NotImplementedException();
    //    }

    //    public Task ProcessRequestsAsync(CancellationToken cancellationToken = default)
    //    {
    //        throw new NotImplementedException();
    //    }

    //    public void Tick(DateTimeOffset now)
    //    {
    //        throw new NotImplementedException();
    //    }

    //    public Task Hander808Parse(ReadOnlySpan<byte> span) {
    //        return Task.CompletedTask;
    //    }
    //    public Task HanderActiveSafetyParse(ReadOnlySpan<byte> span)
    //    {
    //        return Task.CompletedTask;
    //    }
    //}


    //public interface IRequestProcessor
    //{
    //    Task ProcessRequestsAsync(CancellationToken cancellationToken = default);
    //    void Tick(DateTimeOffset now);
    //    void Abort(ConnectionAbortedException ex);
    //}
    //interface IHttpOutputProducer
}